// Package server 压测启动
package server

import (
	"fmt"
	"simqtt-stress/model"
	"simqtt-stress/server/golink"
	"simqtt-stress/server/statistics"
	"strconv"
	"sync"
	"time"
)

const (
	connectionMode = 1 // 1:顺序建立长链接 2:并发建立长链接
)

// Dispose 处理函数
func Dispose(concurrency, totalNumber uint64, request *model.MqttModel) {
	// 设置接收数据缓存
	ch := make(chan *model.RequestResults, 5000)
	var (
		wg          sync.WaitGroup // 发送数据完成
		wgReceiving sync.WaitGroup // 数据处理完成
	)
	wgReceiving.Add(1)
	startPrint := make(chan struct{}, 1)
	go statistics.ReceivingResults(concurrency, ch, &wgReceiving, startPrint)
	tmp := make([]*model.MqttModel, concurrency)
	//wgg := &sync.WaitGroup{}
	for i := uint64(0); i < concurrency; i++ {
		//wgg.Add(1)
		// 连接以后再启动协程
		//go func(ii uint64,wggg *sync.WaitGroup) {
		//	defer wggg.Done()
		//	p := request.New(strconv.Itoa(int(ii)), request.Type)
		//	if e := p.Connect(); e != nil {
		//		fmt.Println("连接失败:", ii, e)
		//	}
		//	tmp[ii] = p
		//}(i,wgg)
		p := request.New(strconv.Itoa(int(i)), request.Type)
		if e := p.Connect(); e != nil {
			fmt.Println("连接失败:", i, e)
		}
		tmp[i] = p
	}
	//wgg.Wait()
	startPrint <- struct{}{} // 通知开始打印
	subChan := make(chan struct{}, 2)
	p := request.New("___", request.Type)
	p.SetAddr("tcp://127.0.0.1:1884")
	e := p.Connect()
	if e != nil {
		panic(e)
	}
	request.SetAddr("tcp://127.0.0.1:1884")
	e = request.Connect()
	if e != nil {
		panic(e)
	}
	request.SubTopic(subChan, request.Topic())
	p.SubTopic(subChan, request.Topic())
	<-subChan
	<-subChan
	time.Sleep(1 * time.Second)
	for i := uint64(0); i < concurrency; i++ {
		wg.Add(1)
		switch request.Type {
		case model.Pub:
			go golink.MqttPub(i, ch, totalNumber, &wg, tmp[i])
		case model.Sub:
			go golink.MqttSub(i, ch, totalNumber, &wg, tmp[i])
		default:
			// 类型不支持
			wg.Done()
		}
	}
	// 等待所有的数据都发送完成
	wg.Wait()
	time.Sleep(2 * time.Second)
	// 延时3秒 确保数据都处理完成了
	close(ch)
	// 数据全部处理完成了
	wgReceiving.Wait()
	num := concurrency * totalNumber * 2
	n := 500
	for {
		time.Sleep(1 * time.Second)
		if model.GetAllRecMsg() >= int64(num) {
			fmt.Println("rec num : ", model.GetAllRecMsg())
			break
		}
		n--
		if n%9 == 0 {
			fmt.Println("now rec num : ", model.GetAllRecMsg())
		}
		if n <= 0 {
			fmt.Println("rec num : ", model.GetAllRecMsg())
			break
		}
	}
	time.Sleep(3 * time.Minute)
	return
}
